package com.joelemon.mockinfluxdbdata.mock.service.insertImpl;

import com.joelemon.mockinfluxdbdata.common.SystemException;
import com.joelemon.mockinfluxdbdata.mock.ConnectConfig;
import com.joelemon.mockinfluxdbdata.mock.DataLine;
import com.joelemon.mockinfluxdbdata.mock.service.DBInsertService;
import com.joelemon.mockinfluxdbdata.utils.OkHttpUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.ListUtils;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.List;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/**
 * @Description: http实现 优化版：实现批量插入
 * @Author: HJY
 * @Date: 2023/11/16
 */
@Slf4j
@Service("httpInsertOptimize")
public class DBHttpOptimizeInsertService implements DBInsertService {
    public static final Predicate<Integer> SUCCESS = i -> i != null && i == 204;

    private final static List<String> protocols = Arrays.asList("http", "https");

    private final static int BATCH_COUNT = 500;

    @Override
    public int insertData(ConnectConfig connectConfig, List<DataLine> dataLines) {
        int count = 0;
        if (CollectionUtils.isEmpty(dataLines)) {
            return count;
        }
        if (!protocols.contains(connectConfig.getProtocol())) {
            throw new SystemException("不支持的协议:" + connectConfig.getProtocol());
        }
        String prefix = connectConfig.getProtocol() + "://";
        String url = prefix + connectConfig.getIp() + ":" + connectConfig.getPort() + "/write?db=" + connectConfig.getDatabase() +
                "&precision=ns&u=" + connectConfig.getUsername() + "&p=" + connectConfig.getPassword() + "&rp=" + connectConfig.getRetentionPolicy();
        // 数据分片
        List<List<DataLine>> partitions;
        if (dataLines.size() > BATCH_COUNT) {
            partitions = ListUtils.partition(dataLines, BATCH_COUNT);
        } else {
            partitions = Arrays.asList(dataLines);
        }
        for (List<DataLine> dataLineList : partitions) {
            String requestData = dataLineList.stream().filter(dataLine -> dataLine.check())
                    .map(dataLine -> {
                        String tagStr = dataLine.getTagMap().entrySet().stream().map(entry -> entry.getKey() + "=" + entry.getValue())
                                .collect(Collectors.joining(","));
                        String filedStr = dataLine.getFieldMap().entrySet().stream().map(entry -> entry.getKey() + "=" + entry.getValue())
                                .collect(Collectors.joining(" "));
                        String dataStr = dataLine.getMeasurement() + "," + tagStr + " " + filedStr + " " + dataLine.getTime().getTime() + "000000";
                        return dataStr;
                    }).collect(Collectors.joining("\n"));
            try {
                log.debug("数据插入：" + requestData);
                String postRawParams = OkHttpUtil.postRawParams(url, requestData, OkHttpUtil.TEXT, SUCCESS, v1 -> v1);
                log.debug("数据插入结果：" + postRawParams);
                count += dataLineList.size();
            } catch (Exception e) {
                log.error("数据插入异常：{}", e);
                e.printStackTrace();
            }
        }
        return count;
    }
}
